package tbdp.tool;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;

import tbdp.tool.common.ColumnDesc;
import tbdp.tool.common.JobConstants;
import tbdp.tool.common.Record;
import tbdp.tool.common.StockInfo;
import tbdp.tool.config.ReaderConfig;
import tbdp.tool.context.JobContext;
import tbdp.tool.spi.Reader;
import tbdp.tool.spi.RecordIterator;
import tbdp.tool.util.VariableContext;

/**
 * @author xbkaishui
 * @version $Id: HbaseRecordIterator.java, v 0.1 2017-06-18 下午12:16 xbkaishui Exp $$
 */
public class HbaseRecordIterator implements RecordIterator<Record> {

    private RowKeyGenerator rowKeyGenerator = new RowKeyGenerator();
    protected Logger logger = LoggerFactory.getLogger(Reader.class);

    private JobContext context;
    private List<StockInfo> stockInfos;
    private HTableInterface table;
    private String tableName;
    private ReaderConfig readerConfig;
    private int curIndex;
    private int total;
    private Iterator<Result> cursor;
    private List<ColumnDesc> cols;

    public HbaseRecordIterator(JobContext context, List<ColumnDesc> cols) {
        this.context = context;
        this.readerConfig = context.getJobConfig().getReader();
        this.tableName = readerConfig.getTableName();
        String tableWithSchema = readerConfig.getTableWithSchema();
        this.stockInfos = (List<StockInfo>) context.getValue(JobConstants.STOCK_LIST);
        this.total = stockInfos.size();
        this.table = Utils.getHbaseTable(tableWithSchema);
        this.cols = cols;

        //roll to next iterator
        getNextIterator();
    }

    @Override
    public boolean hasNext() {
        if (curIndex < total-1) {
            return true;
        }
        if(curIndex == total){
            return false;
        }
        if(curIndex == 90){
            System.out.println("90");
        }
        return cursor.hasNext();
    }

    public boolean isEnd(){
        return curIndex >= total;
    }

    @Override
    public Record next() {
        while (!cursor.hasNext()) {
            if(isEnd()){
                break;
            }
            getNextIterator();
        }
        Result next = cursor.next();
        if(next == null){
            return null;
        }
        Record record = convert(next);
        return record;
    }

    /**
     * result 转换成 Record
     */
    private Record convert(Result r) {
        Record rc = new Record();
        String row = Bytes.toString(r.getRow());
        rc.put("rowkey", row);
        logger.debug("rowkey {} ", row);
        for (ColumnDesc column : cols) {
            String colName = column.getColumn();
            String colType = column.getType();
            List<Cell> vals = r.getColumnCells(Bytes.toBytes("cf1"), Bytes.toBytes(colName));
            if (CollectionUtils.isNotEmpty(vals)) {
                Cell kv = vals.get(0);
                byte[] val = kv.getValue();
                Object colVal = null;
                try {
                    if (colType.equalsIgnoreCase("double")) {
                        colVal = Bytes.toDouble(val);
                    } else if (colType.equalsIgnoreCase("long") || colType.equalsIgnoreCase("int")) {
                        colVal = Bytes.toLong(val);
                    } else {
                        colVal = Bytes.toString(val);
                    }
                } catch (Exception e) {
                    logger.error("convert hbase data type error ", e);
                    colVal = Bytes.toString(val);
                }
                rc.put(colName, colVal);
            }
        }
        return rc;
    }

    public void getNextIterator() {
        Scan scan = new Scan();
        try {
            StockInfo stockInfo = stockInfos.get(curIndex++);
            logger.info("getNextIterator stock " + stockInfo);
            String startRange = readerConfig.getConfigPropAsString(JobConstants.START_RANGE);
            String endRange = readerConfig.getConfigPropAsString(JobConstants.END_RANGE);
            String reverseCode = readerConfig.getConfigPropAsString(JobConstants.REVERSE_CODE);
            String scode = stockInfo.getScode();
            /**
             * 是否reverse code
             */
            if (StringUtils.equalsIgnoreCase(reverseCode, "true")) {
                scode = StringUtils.reverse(scode);
            }
            /**
             * init ctx
             */
            VariableContext ctx = VariableContext.instance();
            ctx.addVariable(Constants.VARI_SCODE, scode);
            ctx.addVariable(Constants.VARI_MARKET_TYPE, stockInfo.getMarketType());

            String startKey = rowKeyGenerator.generateRowKey(tableName, startRange);
            String endKey = rowKeyGenerator.generateRowKey(tableName, endRange);
            logger.info("start rowkey {} idx {} ", startKey ,curIndex );
            logger.info("end rowkey {} idx {} ", endKey,curIndex);
            scan.setStartRow(Bytes.toBytes(startKey));
            scan.setStopRow(Bytes.toBytes(endKey));
            scan.setBatch(Integer.MAX_VALUE);
            ResultScanner resultScanner = table.getScanner(scan);
            cursor = resultScanner.iterator();
        } catch (Throwable e) {
            logger.error("getHbaseData error", e);
        }
    }

}
